package my.thesis.spouts;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.commons.io.IOUtils;
import org.joda.time.DateTime;

import java.net.URL;

import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichSpout;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;

public class TweetReaderUrl extends BaseRichSpout 
{
	private static final long serialVersionUID = 1L;
	private BufferedReader reader;
	private SpoutOutputCollector _collector;
		
	private final String URL = "http://192.168.1.100:80/MY_TWEETS_HASHTAGS_25_10_fortest.json?1";
	private int failedRetries=0;

	//ME ENDIAMESO BUFFER
	private ArrayList<String> toSend = new ArrayList<String>();
	
	public void nextTuple() 
	{
		try 
		{
			//ME BUFFERED READER
			String in;
			in = this.reader.readLine();
			if(in!=null)
			{
					//DEN DOULEVEI!!!
					///System.out.println(in);
					//DateTime timestamp = DateTime.now();
					//long millis = timestamp.getMillis();
					//_collector.emit(new Values(in,millis));
				    
				    //ME ENDIAMESO BUFFER
					toSend.add(in);
			}
			else 
			{
					failedRetries++;
					if(failedRetries > 10)
					{
						System.out.println("End of file reached bb...");
						Thread.sleep(1000000000);	
					}			
			}
			
			for(String send : toSend)
			{								
				DateTime timestamp = DateTime.now();
				long millis = timestamp.getMillis();
				
				_collector.emit(new Values(send,millis)); 
			}
			
			toSend.clear();
			
		}
		
		catch (Exception e)
		{
			System.err.println("Error reading line "+e.getMessage());
		}
	}

	public void open(Map map, TopologyContext context, SpoutOutputCollector outputCollector) 
	{
		this._collector = outputCollector;
		
		try 
		{
			URL url = new URL(URL);
			this.reader = new BufferedReader(new InputStreamReader(url.openStream()));	
		}
		catch (IOException e)
		{
			System.err.println("Error in communication with server ["+e.getMessage()+"]");
		}			
	}

	public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) 
	{
		outputFieldsDeclarer.declare(new Fields("jsonString","timestamp"));	
	}

}
